package gupao.concurrency.datasync.task.v1x;

import gupao.concurrency.datasync.task.AbstractSyncTask;
import gupao.concurrency.utils.FutureUtil;
import gupao.concurrency.utils.JsonUtils;
import gupao.concurrency.utils.ThreadPoolService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

@Slf4j
@Component
public class SyncTaskV1 extends AbstractSyncTask {

     @Autowired
    protected WordProviderMock wordProvider;

    protected final String bizName = "syncTask_v1";

    protected final ThreadPoolExecutor executor = ThreadPoolService.getOrCreate
            (bizName, 4, 100, new SynchronousQueue<>(),
                    new ThreadPoolExecutor.CallerRunsPolicy());

    public void syncStart(int partitionNum, int totalNum) {
        if (start.compareAndSet(false, true)) {
            status = new SyncTaskStatus(new Date());
            new Thread(() -> {
                try {
                    AtomicInteger i = new AtomicInteger(0);
                    long start = System.currentTimeMillis();
                    Collection<List<String>> list = wordProvider.queryAllWord(partitionNum, totalNum);
                    status.setWordProvideTimeCost(System.currentTimeMillis() - start);
                    start = System.currentTimeMillis();
                    List<Future<?>> fsList = list.stream().map(words -> executor.submit(new InnerTask(i.incrementAndGet(), words, status))).collect(Collectors.toList());
                    FutureUtil.waitFuturesComplete(fsList);
                    status.setEndTime(new Date());
                    status.setSyncTimeCost(System.currentTimeMillis() - start);
                    status.setTotalTimeCost(status.getSyncTimeCost() + status.getWordProvideTimeCost());
                    TaskRunResultV1 result = new TaskRunResultV1(executor.getLargestPoolSize(), partitionNum, status);
                    log.info("{}", JsonUtils.toJsonStr(result));
                } finally {
                    status = null;
                    start.set(false);
                }
            }).start();
        } else {
            log.warn("Sync task is running.");
        }
    }

}
